/**在默认情况下采用异步方式向客户端推送消息，也就是说ActiveMQ服务器在向某个消费者会话推送消息后，不会等待消费者的响应信息，直到消费者处理完消息后，
 * 主动向服务端返回处理结果。如果您对自己的消费者性能足够自信，也可以将这个过程设置为“同步”：connectionFactory.setDispatchAsync(false);
 * 默认的策略是ActiveMQ服务端一旦有消息，就主动按照设置的规则推送给当前活动的消费者。其中每次推送都有一定的数量限制，这个限制值就是prefetchSize。
 *     ActiveMQPrefetchPolicy prefetchPolicy = connectionFactory.getPrefetchPolicy();
 *     prefetchPolicy.setQueuePrefetch(50);
 *     connectionFactory.setPrefetchPolicy(prefetchPolicy);
 * prefetchSize默认值如下：
 *     PERSISTENT Message—Queue：prefetchSize=1000
 *     NON_PERSISTENT Message—Queue：prefetchSize=1000
 *     PERSISTENT Message—Topic：prefetchSize=100
 *     NON_PERSISTENT Message—Topic：prefetchSize=32766
 * 1) 同步/异步接收：consumer.receive/consumer.setMessageListener
 * 2）消息事务：在seesion创建时必须启用事务，和发送者保持一致，通过commit来确认收到消息，否则回滚;一组消息要么被全部处理并确认成功，要么全部被回滚并重新处理
 * 3）消息持久化：只有持久订阅者才能收到之前离线期间发送的持久化消息，普通订阅者只能收到在线期间发送的消息
 * 4）消息确认：在收到消息后用session.commit或message.acknowledge进行手动确认
 * 5）死信：默认队列名为ActiveMQ.DLQ
 * 6）延迟发送：在发送时用message.setLongProperty来设置延迟时间
 */

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQMessageConsumer;
import org.apache.activemq.ActiveMQSession;

import javax.jms.*;

public class AdvancedConsumer {

    static QueueConnection connection = null;
    static QueueSession session = null;
    //死信队列可以当成普通队列一样消费
    public static void commonReceiver(String myQueue, Boolean isTranction) {
        try {
            QueueConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
                    ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616");
            connection = connectionFactory.createQueueConnection();
            connection.start();
            session = connection.createQueueSession(isTranction, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue(myQueue);
            QueueReceiver consumer = session.createReceiver(queue);
            consumer.setMessageListener(new QueueListener());

            if (isTranction) session.commit();

            System.in.read();
            consumer.close();
            session.close();
            connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //事务与持久化,一个事务类型的Session中只有一个Consumer，已避免rollback()或者commit()方法被多个consumer调用而造成的消息混乱
    public static void tansactionReceiver() throws JMSException {
        QueueConnection connection = null;
        QueueSession session = null;
        int number = 0;
        try {
            QueueConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
                    ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616");
            connection = connectionFactory.createQueueConnection();
            connection.start();
            //必须保证发送端和接受端都是事务性会话,第二个参数表示使用commit或者rollback的方式进行向服务器应答
            session = connection.createQueueSession(true, Session.SESSION_TRANSACTED);
            Queue queue = session.createQueue("mytranactionqueue");
            QueueReceiver consumer = session.createReceiver(queue);

            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message msg) {
                    if (msg != null) {
                        TextMessage message = (TextMessage) msg;
                        try {
                            System.out.println("[Transaction]received：" + message.getText());
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
            // 休眠2s再关闭
            Thread.sleep(1000 * 2);
            //当前session事务中所有消息立即被确认,此时手动acknowledge不起作用
            if(number++ % 3 == 0) session.commit();
        } catch (Exception e) {
            //将当前事务中的消息清空并要求重发
            session.rollback();
            e.printStackTrace();
        } finally {
            // 关闭释放资源
            if (session != null) {
                session.close();
            }
            if (connection != null) {
                connection.close();
            }
        }
    }


    //若不存在持久订阅者和在线的普通订阅者，这个消息不会保存，当普通订阅者上线后，它是收不到消息的
    //若存在离线的持久订阅者，broker 会为该持久订阅者保存消息，当该持久订阅者上线后，会收到消息
    //持久订阅者/非持久订阅者的区别在于：订阅者离线的期间的消息(包括持久消息和非持久消息)是否能在重新上线后接收到
    public static void durbleReceiver() {
        TopicConnection connection = null;
        TopicSession session = null;
        try {
            TopicConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
                    ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616");
            connection = connectionFactory.createTopicConnection();
            connection.setClientID("myClientID");

            //必须保证发送端和接受端都是事务性会话
            session = connection.createTopicSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
            Topic topic = session.createTopic("mytopic");

            // 使用Session对象创建一个消费者对象。
            MessageConsumer consumer = session.createDurableSubscriber(topic, "dconsumer");
//            TopicSubscriber consumer = session.createDurableSubscriber(topic, "dconsumer");
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message msg) {
                    try {
                        TextMessage textMsg = (TextMessage) msg;
                        System.out.println("[Durable] Received:"
                                + textMsg.getText());
                        msg.acknowledge();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            connection.start();
//            TextMessage msg=(TextMessage) consumer.receive();
//            System.out.println(msg.getText());
            Thread.sleep(1000 * 5);
            session.close();
            connection.close();
        } catch (Exception e) {
            //将当前事务中的消息清空并要求重发
//        session.rollback();
            e.printStackTrace();
        }
    }


    /*ActiveMQ的五种消息确认模式:
     1)AUTO_ACKNOWLEDGE : 自动确认,同步(receive)方法下：optimizeACK未开启则在返回message之前立即确认；optimizeACK开启时，则会在unAck数量达到prefetch * 0.65或时间超过optimizeAcknowledgeTimeOut时批量确认
                                  异步"(messageListener)方法下：如果onMessage方法异常,将导致client端补充发送一个ACK_TYPE为REDELIVERED_ACK_TYPE确认指令；
                                  如果onMessage方法正常,消息将会正常确认(STANDARD_ACK_TYPE)
                                  connectionFactory.setOptimizeAcknowledge(true);connectionFactory.setOptimizeAcknowledgeTimeOut(5000);
     2)CLIENT_ACKNOWLEDGE:手动确认（批量）,通常在消息接受完毕之后确认，也可以在消息被正确处理之后再确认。既可以"逐个"确认消息，也可以在接收或处理多条消息之后一次确认多条消息，减少ack的次数来提升consumer的效率
                          MQ服务器发送个消费者的消息个数count = prefetchExtension + prefetchSize - dispatched.size()，dispatched表示已经发送给client端但是还没有“STANDARD_ACK_TYPE”的消息总量
                          消息在交付给listener之前，都会首先创建一个DELIVERED_ACK_TYPE的ACK指令，直到client端未确认的消息达到"prefetchSize * 0.5"时才会发送此ACK指令，那么此consumer将会被标记为“slow”，此后queue中的消息将会转发给其他Consumer
     3)DUPS_OK_ACKNOWLEDGE:"消息可重复"确认,一种潜在的"AUTO_ACK"确认机制，为批量确认而生，而且具有“延迟”确认的特点
                           如果Destination是Queue，我们真的可以认为DUPS_OK_ACK就是“AUTO_ACK + optimizeACK + (prefetch > 0)”这种情况
                           如果Destination是Topic,无论optimizeACK是否开启，都会在消费的消息个数>=prefetch * 0.5时，批量确认(STANDARD_ACK_TYPE),在此过程中，
                           不会发送DELIVERED_ACK_TYPE的确认指令，这也意味着，当consumer故障重启后，那些尚未ACK的消息会重新发送过来
     4)SESSION_TRANSACTED: 当session使用事务时，就是使用此模式.session.commit()会导致当前session的事务中所有消息立即被确认;session.rollback()意味着当前事务的结束，事务中所有的消息都将被重发,这会导致message.redeliveryCounter计数器增加
     5)INDIVIDUAL_ACKNOWLEDGE: 手动确认（单条）,它的确认时机和CLIENT_ACKNOWLEDGE几乎一样,但是它只确认单条消息，性能很差，除非有特别的业务要求，否则不建议使用。
     */
    public static void ackReceiver() throws JMSException {
        QueueConnection connection = null;
        QueueSession session = null;
        try {
            QueueConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
                    ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616");
            connection = connectionFactory.createQueueConnection();
            connection.start();
            //必须保证发送端和接受端都是事务性会话
            session = connection.createQueueSession(Boolean.FALSE, 4);
            Queue queue = session.createQueue("myackqueue");
            MessageConsumer consumer = session.createConsumer(queue);
            int cnt = 1;
            do {
                // 接收数据的时间（等待） 100 ms
                TextMessage message = (TextMessage) consumer.receive();
                if (message == null) {
                    break;
                }
                System.out.println("[Ack]received：" + message.getText());

                //手动确认消息
                if (cnt % 2 == 0) {
                    //将当前session中所有consumer中尚未确认的消息一起确认
                    message.acknowledge();    //等效：ActiveMQSession.acknowledge();
                    System.out.println("Acknowledge----" + cnt);
                }
                cnt++;
            } while (cnt <= 5);

//            consumer.setMessageListener(new MessageListener() {
//                int msgCount=0;
//                public void onMessage(Message msg) {
//                    if (msg != null) {
//                        msgCount++;
//                        TextMessage message = (TextMessage) msg;
//                        try {
//                            System.out.println("[ACK]received："+message.getText()+msgCount);
//                            //手动确认消息
//                            if (msgCount%2 == 0) {message.acknowledge();
//                                System.out.println("ddddd");}
//                        } catch (JMSException e) {
//                            e.printStackTrace();
//                        }
//                    }
//                }
//            });
            // 休眠2s再关闭
            Thread.sleep(1000 * 2);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭释放资源
            if (session != null) {
                session.close();
            }
            if (connection != null) {
                connection.close();
            }
        }

    }
}


